/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.benchmark;

import java.io.File;
import java.util.Properties;
import java.util.StringTokenizer;

import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.qos.ConsumerConstraints;
import cgl.narada.util.ConfigurationParamsReader;

/** 
    This is for multiple connections, publishers and subscribers.

    @author Shrideep Pallickara
    $Revision$
    $Date$
*/


public class GaugeMultipleSubscriberCase {
  private int numOfConnections;
  private int numOfSessionsPerConnection;
  private int numOfSubscribersPerSession;
  private int numOfMessagesPerTestCase;
  private int[] messageSizes;

  /** The total number of subscribers in the system is
      numOfSubscribersPerSession x numOfSessionsPerConnection x 
      numOfConnections */
  private ClientService[] clientServices;
  
  private boolean createMeasuringClient=true;
  private boolean consoleOutput=false;

  private GaugeSubscriber measuringClient;
  private int maxReject=0;
  private String moduleName = "GaugeMultipleSubscriberCase: ";

  public 
  GaugeMultipleSubscriberCase(int _numConnections, int _numSubscribersPerSess,
			      int _numOfMessagesPerTestCase, 
			      int[] messageSizes) {
    numOfConnections = _numConnections;
    numOfSubscribersPerSession = _numSubscribersPerSess;
    numOfMessagesPerTestCase = _numOfMessagesPerTestCase;
    this.messageSizes = messageSizes;

    clientServices = new ClientService[numOfConnections];

    int numOfSessions = numOfConnections*numOfSessionsPerConnection;
    
    int numOfSubscribers =
      numOfSessions*numOfSubscribersPerSession;
    

  }
  

  public void createMeasuringClient(boolean create, int maxRejects) {
    createMeasuringClient = create;
    maxReject = maxRejects;
  }
  
  public void printToConsole(boolean console) {
    consoleOutput = console;
  }

  public void initialize(int entityId, Properties props, String commType) 
    throws ServiceException {

    for (int i=0; i< numOfConnections; i++) {
      clientServices[i] = SessionService.getClientService(entityId++);
      props.put("UDPListenerPort", ""+entityId);
      clientServices[i].initializeBrokerCommunications(props, commType);      
    }/* end - for */    

  }
  
  public void
  initializeSessionsAndSubscribers(int profileType, Object subscription,
				   Properties subProperties, int templateId, 
				   String reportFilename)
    throws ServiceException {
    int subscriberId=0;
    boolean receiveReliably = false;
    if (subProperties.getProperty("ReliableDelivery").equals("true") ) {
      System.out.println("\n\n" + moduleName + "Subscribers are reliable!\n");
      receiveReliably = true;
    } else {
      System.out.println("\n\n" + moduleName + "NORMAL subscriber");
    }
    
    for (int i=0; i < numOfConnections; i++) {
      ClientService _clientService = clientServices[i];
      
      for (int k=0; k < numOfSubscribersPerSession; k++) {
	Profile _profile = 
	  _clientService.createProfile(profileType, subscription);
	 
	GaugeSubscriber _gaugeSub =  new GaugeSubscriber(++subscriberId); 
	EventConsumer _consumer= _clientService.createEventConsumer(_gaugeSub);
	
	if (receiveReliably) {
	  ConsumerConstraints _constraints = 
	    _consumer.createConsumerConstraints(_profile);
	  _constraints.setReceiveReliably(templateId);
	  _consumer.subscribeTo(_profile, _constraints);
	  long recoveryId = _consumer.recover(templateId, _gaugeSub);
	} else {
	  _consumer.subscribeTo(_profile);
	}

	if (createMeasuringClient) {
	  
	  if (i+k == 0) {
	    measuringClient = _gaugeSub;
	    if (consoleOutput) {
	      measuringClient.printToConsole(consoleOutput);
	    } /*end consoleOutput */
	    
	    measuringClient.setAsMeasuringClient(reportFilename, 
						 numOfMessagesPerTestCase,
						 messageSizes, maxReject);
	  } /*end (i+k == 0) */
	  
	}/** createMeasuringClient */
	
      }/** loop(k)*/
      
    }/** loop(i) */
  }
  
  
  public static void main(String[] args) {
    String module = "GaugeMultipleSubscriberCase.main()->" ;
    String gaugeParamsFile = "GaugeParams.txt";
    if ( (args.length < 2) || (! args[0].equals("-f")) ) {
      System.out.println(module + "Usage Options: \n" + 
			 "FOR SINGLE MEASURING Subscriber =>\n" + 
			 "java cgl.narada.benchmark." +
			 "GaugeMultipleSubscriberCase -f <gaugeParams> \n" +
			 "\t\t\t\tOR\n" + "FOR MULTIPLE Non-Measuring " +
			 "Subscribers =>\n" + "java cgl.narada.benchmark." +
			 "GaugeMultipleSubscriberCase -f <gaugeParams> " + 
			 "<someText>");
      System.exit(0);
    } else {
      gaugeParamsFile=args[1];
    }
    
    System.out.println(module + "Using file [" +  gaugeParamsFile + 
		       "] for initialization of benchmark.");
    
    
    ConfigurationParamsReader bmProperties = 
      new ConfigurationParamsReader(gaugeParamsFile);
    
    int numConnections = 
      Integer.parseInt(bmProperties.getProperty("NumberOfConnections"));
    
    int numSubscribersPerSession = 
      Integer.parseInt(bmProperties.getProperty("NumberOfSubscribersPerSession"));
    
    
    String val = bmProperties.getProperty("CONSOLE-OUTPUT");
    
    boolean consoleOutput = false;
    if (val.equals("true"))
      consoleOutput=true;
    
    int numOfMessages = 
      Integer.parseInt(bmProperties.getProperty("NumOfMessagesPerTestCase"));

    String sizes = bmProperties.getProperty("MessageSizes");
    System.out.println("Message sizes =" + sizes);
    StringTokenizer tok = new StringTokenizer(sizes, ",[] ");
    int numTokens = tok.countTokens();
    int[] messageSizes = new int[numTokens];
    for (int i=0; i<numTokens; i++) {
      messageSizes[i] = Integer.parseInt(tok.nextToken().trim());
    }

    int maxReject = Integer.parseInt(bmProperties.getProperty("MaxReject"));

    boolean createMeasuringClient = true;
    String reportFilename="MeasuringSubscriberReportXXX.txt";
    if (args.length > 2) {
      createMeasuringClient = false;
    } else {
       System.out.println("\n" + module + 
			  "Measuring Subscriber to be instantiated \n");
       
    }
    
    if (createMeasuringClient) {
      numConnections = 1;
      numSubscribersPerSession = 1;
      String reportDirectory = bmProperties.getProperty("ReportsDirectory");
      File reportDir = new File(reportDirectory);
      String reportDirLocation = reportDir.getAbsolutePath();
      if ( (!reportDir.exists()) || (!reportDir.isDirectory())   ) {
	System.out.println(module + "The specified reports directory"
			   + "[" + reportDirLocation + "] does not exist"
			   + ". Please check this.");
	System.exit(0);
      }
      reportFilename = reportDirLocation + "/" +
	bmProperties.getProperty("MeasuringSubscriberReportFile");
    }
    
    String brokerHost = bmProperties.getProperty("BROKER-HOST");
    String brokerPort= bmProperties.getProperty("BROKER-PORT");
    
    Properties props = new Properties();
    props.put("hostname", brokerHost); 
    props.put("portnum", brokerPort);
    String commType =  bmProperties.getProperty("COMM-TYPE");
    int entityId = Integer.parseInt(bmProperties.getProperty("SubEntityId"));
    int templateId = Integer.parseInt(bmProperties.getProperty("TemplateId"));

    System.out.println(module + "Read templateId");
    int profileType = TemplateProfileAndSynopsisTypes.STRING;
    Object subscription = "Test/benchmark";
    String reportFile = "Report";

    GaugeMultipleSubscriberCase gmsc = 
      new GaugeMultipleSubscriberCase(numConnections, numSubscribersPerSession,
				      numOfMessages, messageSizes);
    gmsc.printToConsole(consoleOutput);
    gmsc.createMeasuringClient(createMeasuringClient, maxReject);
    
    Properties subscriberProperties = new Properties();
    
    String reliableDelivery = bmProperties.getProperty("ReliableDelivery");

    if (reliableDelivery.equals("true") ) {
      subscriberProperties.setProperty("ReliableDelivery", "true");
    } else {
      subscriberProperties.setProperty("ReliableDelivery", "false");
    }

    try {
      gmsc.initialize(entityId, props, commType);    
      gmsc.initializeSessionsAndSubscribers(profileType, subscription,
					    subscriberProperties, templateId,
					    reportFilename);
    } catch (ServiceException serEx) {
      System.out.println(module + "Error while performing benchmarks " +
			 serEx);
    }
    
    
  }
}  
